package com.gjy.kafka.java;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * 1. 实现Partitioner
 * 2. 实现3个方法: partition、close、configure
 * 3. 编写partition方法,返回分区号
 *
 * @author gjy
 * @version 1.0
 * @since 2024-06-23 18:10:57
 */
public class PartitionerImpl implements Partitioner {

    private static final Logger log = LoggerFactory.getLogger(PartitionerImpl.class);

    /**
     * @param topic     主题
     * @param key       消息的key
     * @param keyByte   key序列化数据
     * @param value     value
     * @param valueByte value序列化数组
     * @param cluster   集群元数据可以查看分区信息
     * @return 分区号
     */
    @Override
    public int partition(String topic, Object key, byte[] keyByte,
                         Object value, byte[] valueByte, Cluster cluster) {
        String msg = value.toString();
        int partition;
        if (msg.contains("p0")) {
            partition = 0;
        } else if (msg.contains("p1")) {
            partition = 1;
        } else {
            partition = 2;
        }
        return partition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
